﻿#include <stdarg.h>
#include <unistd.h>
#include "MemoryPoolObj.h"

#include "define_protocol.h"

#include "SendThread.h"


CSendThread::CSendThread()
{
    m_iSendMsgQueueCount=0;
    m_iThreadNum = 1;
    if (pthread_mutex_init(&m_sendMessageQueueMutex, nullptr) != 0)
    {
        printf("m_sendMessageQueueMutex  Init error \n");
    }
    if (sem_init(&m_semEventSendQueue, 0, 0) == -1)
    {
        printf("m_semEventSendQueue Init error \n");
    }
    m_shutdown = false;
}

CSendThread::~CSendThread()
{    
    ClearMsgSendQueue();
}

void CSendThread::ClearMsgSendQueue()
{
	char * sTmpMempoint=nullptr;
	while(!m_MsgSendQueue.empty())
	{
		sTmpMempoint = m_MsgSendQueue.front();
        m_MsgSendQueue.pop_front();
        MemPoolObj::GetMemoryPool()->Free((void*)sTmpMempoint); 
        sTmpMempoint = nullptr;
	}	
}
bool CSendThread::Start(int nID)
{    
 
    //m_iThreadNum = nID;
    
    for(int i = 0; i < 1; ++i)
    {
		ThreadItemSend *pNew = new ThreadItemSend(this);
        m_threadVector.push_back(pNew);             //创建 一个新线程对象 并入到容器中         
        int err = pthread_create(&pNew->_Handle, nullptr, ThreadSendFun, pNew);      //创建线程，错误不返回到errno，一般返回错误码
        if(err != 0)
        {
            
            return false;
        }
        else
        {
            
        }        
    } 
    std::vector<ThreadItemSend*>::iterator iter;
lblfor:
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        if( (*iter)->ifrunning == false)
        {
            
            usleep(100 * 1000);
            goto lblfor;
        }
    }
    return true;
    printf("CSendThread Start ----------->\n");

}

void CSendThread::Stop() 
{
    printf("CSendThread StopAll begin..\n");
    while (m_iSendMsgQueueCount>0)
    {
        //printf("wait: m_iSendMsgQueueCount =%d\n", (int)m_iSendMsgQueueCount);
        sem_post(&m_semEventSendQueue);
        sleep(1);
    }
    
    if(m_shutdown == true)
    {
        return;
    }
    m_shutdown = true;
   
    printf("CSendThread StopAll m_shutdown =true\n");


    int nDex = 0;
	for (nDex = 0; nDex < 1; nDex++)
	{
		sem_post(&m_semEventSendQueue);
		printf("stop sem_post...\n");
	}
    printf("CSendThread StopAll sem_post: %d\n", (int)m_iSendMsgQueueCount);
	
    std::vector<ThreadItemSend*>::iterator iter;
	
	for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        pthread_join((*iter)->_Handle, nullptr); //等待一个线程终止
    }
	
    printf("CSendThread StopAll pthread_join\n");

	for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
	{
		if(*iter)
		{
			delete *iter;
		}
	}
	m_threadVector.clear();
    
    return;    
}

void CSendThread::AddTask(char* psendbuf)
{
    
    pthread_mutex_lock(&m_sendMessageQueueMutex); 

    
    LPSTRUC_MSG_HEADER pMsgHeader = (LPSTRUC_MSG_HEADER)psendbuf;
	
    sClient *p_Conn = pMsgHeader->p_Conn;
    if (p_Conn->nSendCount > 50)
    {
        return;
    }
    ++p_Conn->nSendCount;                                                                //发送队列中有的数据条目数+1；
    m_MsgSendQueue.push_back(psendbuf);
    ++m_iSendMsgQueueCount;                                                         //原子操作

    //printf("CSendThread:m_iSendMsgQueueCount:%d \n",(int)m_iSendMsgQueueCount);

    pthread_mutex_unlock(&m_sendMessageQueueMutex);
    
    
    if (sem_post(&m_semEventSendQueue) == -1) 
    {
        printf("sem_post eror \n");
    }
    //printf("sem_post  end\n");
    return;
}

void* CSendThread::ThreadSendFun(void* threadData)
{


    ThreadItemSend * pThread = static_cast<ThreadItemSend*>(threadData);
	
    CSendThread* pThis = pThread->_pThis;
	
    
    

    char* pMsgBuf;
    LPSTRUC_MSG_HEADER	pMsgHeader;
    LPCOMM_PKG_HEADER   pPkgHeader;
    sClient  *p_Conn;
    unsigned short      itmp;
    ssize_t             sendsize;
    printf("CSendThread::ThreadSendFun begin Size:%d \n", (int )pThis->m_iSendMsgQueueCount);
    while (1) 
    {
       
       if (pThread->ifrunning == false)
        {
            pThread->ifrunning = true;
        }
       //printf("sem_wait Send  thread begin Size:%d \n", (int)pThis->m_iSendMsgQueueCount);
        if (sem_wait(&pThis->m_semEventSendQueue) == -1)
        {
            if (errno != EINTR) 
            {
               printf("CSocekt::ServerSendQueueThread()中sem_wait(&pThis->m_semEventSendQueue)失败 \n");
            }
        }
       // printf("sem_wait Send thread end Size:%d \n" ,(int)pThis->m_iSendMsgQueueCount);
        
        if (pThis->m_shutdown == true)
        {
            break;
        }
       // printf("CSendThread::ThreadSendFun 22222 \n");
        if (pThis->m_iSendMsgQueueCount > 0) //原子的 
        {
            std::list <char*>::iterator pos, posend;
			
            int err = pthread_mutex_lock(&pThis->m_sendMessageQueueMutex); //因为我们要操作发送消息对列m_MsgSendQueue，所以这里要临界            
            if (err != 0)
            {
               printf("CSocekt::ServerSendQueueThread()中pthread_mutex_lock()失败，返回的错误码为%d \n", err);
            }
            
            pos     = pThis->m_MsgSendQueue.begin();
            posend  = pThis->m_MsgSendQueue.end();
            while (pos != posend)
            {
                pMsgBuf = (*pos);                                                                       //拿到的每个消息都是 消息头+包头+包体【但要注意，我们是不发送消息头给客户端的】
                pMsgHeader = (LPSTRUC_MSG_HEADER)pMsgBuf;   //指向消息头
                pPkgHeader = (LPCOMM_PKG_HEADER)(pMsgBuf + g_TcpMsgSize);	//指向包头
                p_Conn = pMsgHeader->p_Conn;
                                
                //包过期,断开，已回收；
                if (p_Conn->iCurrsequence != pMsgHeader->iCurrsequence)
                {
                    pos=pThis->m_MsgSendQueue.erase(pos);
                    
                    --p_Conn->nSendCount;
                    
                    --pThis->m_iSendMsgQueueCount; 			//发送消息队列容量少1

                    MemPoolObj::GetMemoryPool()->Free((void*)pMsgBuf);
                    pMsgBuf = nullptr;
                    //printf("CloseConnection Have  Call  Give up packet:%d,%d  closed:%d \n", (int)p_Conn->nSendCount, (int)p_Conn->nRecvCount, (int)p_Conn->m_bClosed);                    
                    //g_pServer->CloseConnection(p_Conn);
                    continue;
                } 

                if (p_Conn->iThrowsendCount > 0)
                {              
                    pos++;                  
                    printf("continue满:p_Conn->iThrowsendCount : %d  SendQueueSize:%d \n", (int)p_Conn->iThrowsendCount, (int)pThis->m_iSendMsgQueueCount); //做个提示吧，商用时可以干掉
                    continue;
                }
                //sleep(5);
              
                
                p_Conn->psendMemPointer = pMsgBuf;      //发送后释放用的，因为这段内存是new出来的

                pos=pThis->m_MsgSendQueue.erase(pos);

                --p_Conn->nSendCount;

                --pThis->m_iSendMsgQueueCount;              //发送消息队列容量少1	

                p_Conn->psendbuf = (char*)pPkgHeader;    //要发送的数据的缓冲区指针，因为发送数据不一定全部都能发送出去，我们要记录数据发送到了哪里，需要知道下次数据从哪里开始发送

                itmp = ntohs(pPkgHeader->sDataLen) +g_TcpHeaderSize ;             //包头+包体 长度 ，打包时用了htons【本机序转网络序】，所以这里为了得到该数值，用了个ntohs【网络序转本机序】；
                
                p_Conn->isendlen = itmp;                                //要发送多少数据，因为发送数据不一定全部都能发送出去，我们需要知道剩余有多少数据还没发送

                sendsize = pThis->SendProc(p_Conn, p_Conn->psendbuf, p_Conn->isendlen); //注意参数
                //printf("Send  Size:%d  \n", sendsize);

                if (sendsize > 0)
                {
                    if (sendsize == p_Conn->isendlen) 
                    {
                        
                        MemPoolObj::GetMemoryPool()->Free((void*)pMsgBuf);//....dError
                        p_Conn->psendMemPointer = nullptr;
                        p_Conn->iThrowsendCount = 0;         //这行其实可以没有，因此此时此刻这东西就是=0的
                        //printf("SendProc发送完毕 %d  \n", sendsize);
                        //printf("CloseConnection 00000: %d %d closed:%d \n", (int)p_Conn->nSendCount, (int)p_Conn->nRecvCount, (int)p_Conn->m_bClosed);
                        g_pServer->CloseConnection(p_Conn);  //dError                                             
                    }
                    else 
                    {
                        
                        p_Conn->psendbuf = p_Conn->psendbuf + sendsize;
                        p_Conn->isendlen = p_Conn->isendlen - sendsize;                      
                        ++p_Conn->iThrowsendCount;
                        g_pServer->UpdateEvent(p_Conn, p_Conn->fd, EPOLLOUT,0);
                        printf("11数据没发送完毕【发送缓冲区满】，整个要发送%d，实际发送了%d 剩余:%d p_Conn->iThrowsendCount : %d QueueSize:%d\n", (int)itmp, (int)sendsize, (int)(itmp-sendsize), (int)p_Conn->iThrowsendCount, (int)pThis->m_iSendMsgQueueCount);
                    }
                    continue;                     
                }  
                
                else if (sendsize == 0)
                {
                    MemPoolObj::GetMemoryPool()->Free((void*)pMsgBuf);
                    p_Conn->psendMemPointer = nullptr;
                    p_Conn->iThrowsendCount = 0;    
                    
                  //  printf("CloseConnection 44: %d %d closed:%d \n", (int)p_Conn->nSendCount, (int)p_Conn->nRecvCount, (int)p_Conn->m_bClosed);
                    g_pServer->CloseConnection(p_Conn);  //centos7.6   注释
                    continue;
                }
                //能走到这里，继续处理问题
                else if (sendsize == -1)
                {
                    
                    ++p_Conn->iThrowsendCount;                                                              //标记发送缓冲区满了，需要通过epoll事件来驱动消息的继续发送
                    g_pServer->UpdateEvent(p_Conn, p_Conn->fd, EPOLLOUT,0);
                    sem_post(&pThis->m_semEventSendQueue);
                    printf("22数据没发送完毕【发送缓冲区满】，整个要发送%d，实际发送了%d p_Conn->iThrowsendCount : %d  QueueSize:%d\n", p_Conn->isendlen, sendsize, (int)p_Conn->iThrowsendCount, (int)pThis->m_iSendMsgQueueCount);
                    continue;
                }

                else
                {
                    //能走到这里的，应该就是返回值-2了，一般就认为对端断开了，等待recv()来做断开socket以及回收资源        
                    MemPoolObj::GetMemoryPool()->Free((void*)pMsgBuf);
                    p_Conn->psendMemPointer = nullptr;
                    p_Conn->iThrowsendCount = 0;  //这行其实可以没有，因此此时此刻这东西就是=0的  
                    //printf("CloseConnection 55: %d %d closed:%d \n", (int)p_Conn->nSendCount, (int)p_Conn->nRecvCount, (int)p_Conn->m_bClosed);
                    g_pServer->CloseConnection(p_Conn);  //centos7.6   注释
                    continue;
                }

            } 
            err = pthread_mutex_unlock(&pThis->m_sendMessageQueueMutex);
            if (err != 0)
            {
                printf( "CSocekt::ServerSendQueueThread()pthread_mutex_unlock()失败，返回的错误码为%d! \n", err);
            }
        }

    } 

    printf("CSendThread::ThreadSendFun exit \n");
    return (void*) 0;
}
ssize_t CSendThread::SendProc(sClient *c, char* buff, ssize_t size)
{
    ssize_t   n;
    while(1)
    {
        CAutoLockMe autoLockDevice(c->m_bClosedLock);
		if(c->m_bClosed == true)
		{
			return  -2;
		}
        n = send(c->fd, buff, size, 0); //send()系统函数， 最后一个参数flag，一般为0； 
        if (n > 0)              //成功发送了一些数据
        {
            return n;           //返回本次发送的字节数
        }
        if (n == 0)             //对方断开，不在这里处理；
        {
            return 0;
        }
        if (errno == EAGAIN)    //这东西应该等于EWOULDBLOCK
        {
            return -1;          //表示发送缓冲区满了
        }
        if (errno == EINTR)     //被信号中断
        {
			continue;
        }
        else
        {         
            return -2;
        }
    }
    return 0;
}


